package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"6.5840/labgob"
	"bytes"
	//	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	//	"6.5840/labgob"
	"6.5840/labrpc"
)

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type Entry struct {
	//Index   int
	Term    int
	Command interface{}
}

// 服务器状态分为三种：follower、leader、candidate
const (
	Follower = iota
	Candidate
	Leader
)

const (
	HeartbeatTimeout = 101
	ElectionTimeout  = 450
)

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// Persistent state on all servers
	currentTerm int     // latest term has seen，服务器中的一个逻辑时间戳
	votedFor    int     // candidateId that received vote in current term(or null if none)
	log         []Entry // log entries: each entry contains command for state machine. and term when entry was received by leader(first is 1)

	// Volatile state on leaders(Reinitialized after election)
	// nextIndex[]：Leader认为下一个追加日志在每个节点（peer）的索引
	// matchIndex[]：Leader认为每个节点中已经复制的日志项的最高索引
	nextIndex  []int // for each server, index of the next log entry to send to that server(initialed to leader last log index + 1)
	matchIndex []int // index of highest log entry known to be replicated on server.(initialed to 0. increase monotonically)

	// Volatile state on all servers
	// commitIndex：已知的最高日志entry的索引
	// lastApplied：应用于状态机的最高日志的索引
	commitIndex int // index of highest log entry known to be commited.(initialed to 0. increase monotonically)
	lastApplied int // index of highest log entry applied to state machine.(initialed to 0. increase monotonically)

	// new field
	voteTimer      *time.Timer // 为什么记录时间戳而不是使用go语言中的定时器time.Timer（因为不好用）
	heartbeatTimer *time.Timer
	randNum        *rand.Rand
	state          int // 记录当前服务器的身份信息

	applych   chan ApplyMsg
	condApply *sync.Cond

	// 2D
	snapshot          []byte // 快照
	lastIncludedIndex int    // 快照日志中最高索引
	lastIncludedTerm  int    // 快照日志中最高索引对应的term
}

func (rf *Raft) Print() {
	DPrintf("raft%v:{currentTerm=%v, role=%v, }", rf.me, rf.currentTerm, rf.state)
}

// 重置超时时间为[450.0ms, 600.0ms)，当定时器到期时触发重新选举的操作
func (rf *Raft) ResetVoteTimer() {
	rdTimeOut := GetRandomElectTimeOut(rf.randNum)
	rf.voteTimer.Reset(time.Duration(rdTimeOut) * time.Millisecond)
}

// 设置发送心跳信息的时间间隔为101ms，以满足1s内发送的心跳不超过10次
func (rf *Raft) ResetHeartTimer(timeStamp int) {
	rf.heartbeatTimer.Reset(time.Duration(timeStamp) * time.Millisecond)
}

// 计算出整个日志entries的全局递增索引，即虚拟索引
func (rf *Raft) VirtualLogIndex(rIndex int) int {
	return rIndex + rf.lastIncludedIndex
}

// 计算出相对于快照的索引，用于访问log的真实下标，即真实索引
func (rf *Raft) RealLogIndex(vIndex int) int {
	return vIndex - rf.lastIncludedIndex
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	// 获取当前server的current Term以及当前服务器是否为leader
	//var term int
	//var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	return rf.currentTerm, rf.state == Leader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
// before you've implemented snapshots, you should pass nil as the
// second argument to persister.Save().
// after you've implemented snapshots, pass the current snapshot
// (or nil if there's not yet a snapshot).
func (rf *Raft) persist() {
	// Your code here (2C).

	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	// 2C
	e.Encode(rf.votedFor)
	e.Encode(rf.currentTerm)
	e.Encode(rf.log)
	// 2D
	e.Encode(rf.lastIncludedIndex)
	e.Encode(rf.lastIncludedTerm)

	raftstate := w.Bytes()
	rf.persister.Save(raftstate, rf.snapshot) // 2C不用持久化snapshot，2D进行持久化

}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) == 0 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)

	var votedFor int
	var currentTerm int
	var log []Entry
	var lastIncludedIndex int
	var lastIncludedTerm int

	if d.Decode(&votedFor) != nil ||
		d.Decode(&currentTerm) != nil ||
		d.Decode(&log) != nil ||
		d.Decode(&lastIncludedIndex) != nil ||
		d.Decode(&lastIncludedTerm) != nil {
		DPrintf("server %v read persister failed for votedFor=%v, currentTerm=%v", rf.me, votedFor, currentTerm)
	} else {
		// 2C
		rf.votedFor = votedFor
		rf.currentTerm = currentTerm
		rf.log = log
		// 2D
		rf.lastIncludedIndex = lastIncludedIndex
		rf.lastIncludedTerm = lastIncludedTerm

		rf.commitIndex = lastIncludedIndex
		rf.lastApplied = lastIncludedIndex

	}
}

func (rf *Raft) readSnapshot(data []byte) {
	if len(data) == 0 {
		DPrintf("server %v read snapshot failed for %v", rf.me, rf.currentTerm)
		return
	}

	rf.snapshot = data
	DPrintf("server %v read snapshot successfully", rf.me)
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	if rf.commitIndex < index || index <= rf.lastIncludedIndex {
		// index大于commitIndex，说明可能包含未提交的日志，不能加入快照
		// index小于等于lastIncludedIndex，说明可能是重复或更旧的快照
		return
	}

	rf.snapshot = snapshot

	rf.lastIncludedTerm = rf.log[rf.RealLogIndex(index)].Term
	// 截断log，index处的log被存在索引为0的地方
	rf.log = rf.log[rf.RealLogIndex(index):]
	rf.lastIncludedIndex = index
	// 注意此处的bug，应该先截断日志，再把截断处的索引赋值（因为我搞反了导致从上次快照位置）
	//rf.lastIncludedIndex = index
	//rf.log = rf.log[rf.RealLogIndex(index):]
	if rf.lastApplied < index {
		rf.lastApplied = index
	}

	rf.persist()
}

// TODO: InstallSnapshot RPCs are initialed by leader to send chunks of a snapshot to a follower
type InstallSnapshotArgs struct {
	Term              int    // leader's term
	LeaderId          int    // follower can redirect clients
	LastIncludedIndex int    // the snapshot replaces all entries up through and including this index(日志中快照替换的最后一个entry的索引)
	LastIncludedTerm  int    // term of lastIncludedIndex
	Data              []byte // raw bytes of the snapshot chunk, starting at offset（本实验不使用offset）

	LastIncludedCmd interface{} // 自己新加的字段, 用于在0处占位
}

type InstallSnapshotReply struct {
	Term int // currentTerm for leader to update itself
}

func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	// 1.Reply immediately if term < currentTerm
	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm

		return
	}
	// 不需要使用offset机制去分割snapshot，因此2、3、4、5都跳过

	if args.Term > rf.currentTerm {
		// 接收来自新leader的信息，更改自身状态但不影响快照的安装
		rf.currentTerm = args.Term
		rf.votedFor = -1
	}

	rf.state = Follower
	rf.ResetVoteTimer()

	if args.LastIncludedIndex < rf.lastIncludedIndex || args.LastIncludedIndex < rf.commitIndex {
		// 快照比当前持有的快照更旧，不用安装快照
		// 快照比commit的时间更旧，更不用安装快照
		reply.Term = rf.currentTerm
		return
	}

	// 6.if existing log entry has same index and term as snapshot's last included entry, retain log entries following it and replay
	flag := false
	ridx := 0
	for ; ridx < len(rf.log); ridx++ {
		// 找到具有相同index和Term的log entry，截断前面的log保留后面的log
		if rf.VirtualLogIndex(ridx) == args.LastIncludedIndex && rf.log[ridx].Term == args.LastIncludedTerm {
			flag = true
			break
		}
	}

	msg := &ApplyMsg{
		SnapshotValid: true,
		Snapshot:      args.Data,
		SnapshotTerm:  args.LastIncludedTerm,
		SnapshotIndex: args.LastIncludedIndex,
	}
	// 7.Discard the entire log
	if flag {
		rf.log = rf.log[ridx:]
	} else {
		rf.log = make([]Entry, 0)
		rf.log = append(rf.log, Entry{Term: rf.lastIncludedTerm, Command: args.LastIncludedCmd})
	}

	// 8. Reset state machine using snapshot contents (and load snapshot’s cluster configuration)
	rf.snapshot = args.Data // 更新状态机上的内容出错，已修改
	rf.lastIncludedIndex = args.LastIncludedIndex
	rf.lastIncludedTerm = args.LastIncludedTerm

	if rf.commitIndex < args.LastIncludedIndex {
		rf.commitIndex = args.LastIncludedIndex
	}
	if rf.lastApplied < args.LastIncludedIndex {
		rf.lastApplied = args.LastIncludedIndex
	}

	reply.Term = rf.currentTerm
	rf.applych <- *msg
	rf.persist()
}

func (rf *Raft) handleInstallSnapshot(server int) {
	reply := &InstallSnapshotReply{}

	rf.mu.Lock()

	if rf.state != Leader {
		rf.mu.Unlock()
		return
	}

	args := &InstallSnapshotArgs{
		Term:              rf.currentTerm,
		LeaderId:          rf.me,
		LastIncludedIndex: rf.lastIncludedIndex,
		LastIncludedTerm:  rf.lastIncludedTerm,
		Data:              rf.snapshot,
		LastIncludedCmd:   rf.log[0].Command,
	}

	rf.mu.Unlock()

	ok := rf.sendInstallSnapshot(server, args, reply)
	if !ok {
		return
	}

	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	if rf.state != Leader || rf.currentTerm != args.Term {
		return
	}

	if reply.Term > rf.currentTerm {
		rf.currentTerm = reply.Term
		rf.votedFor = -1
		rf.state = Follower
		rf.ResetVoteTimer()
		rf.persist()
		return
	}

	// InstallSnapshot RPC成功发送，说明peer已更新快照，因此将leader中的matchIndex和nextIndex更新
	if rf.matchIndex[server] < args.LastIncludedIndex {
		rf.matchIndex[server] = args.LastIncludedIndex
	}
	rf.nextIndex[server] = rf.matchIndex[server] + 1
}

// TODO: AppendEntries RPCs are initialed by leader to replicated log entries and to provide a form of heartbeat

// AppendEntries RPC Arguments
type AppendEntriesArgs struct {
	Term         int     // leader's term
	LeaderId     int     // follower can redirect clients，follower需要把信息重定向给leader
	PrevLogIndex int     // index of log entry immediately preceding new ones，追加新日志前的日志索引
	PrevLogTerm  int     // term of prevLogIndex entry，追加新日志前的日志的Term
	Entries      []Entry // log entries to store(empty for heartbeat; may send more than one for efficiency)
	LeaderCommit int     // leader's commitIndex，leader记录的已经commit的日志项的最高索引
}

// AppendEntries RPC Reply
type AppendEntriesReply struct {
	Term    int  // currentTerm, for leader to update itself
	Success bool // true if follower contained entry matching prevLogIndex and prevLogTerm

	// new field
	XTerm  int // follower conflict with leader log's term
	XIndex int // follower conflict log's first index
	XLen   int // follower log's length
}

// leader发送的RPC，用于更新log和心跳消息
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	// Your data here (2A, 2B).
	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	if args.Term < rf.currentTerm {
		// 1.Reply false if term < currentTerm($5.1)
		// 分为两种情况
		// (1)收到了来自旧leader的信息
		// (2)当前节点是孤立节点，一直再持续增长进行自我选举，因此真正的leader发送的消息是旧的term
		reply.Term = rf.currentTerm
		reply.Success = false
		DPrintf("server %v receive old leader% v heartbeat, args=%+v, update term: %v\n",
			rf.me, args.LeaderId, args, reply.Term)
		return
	}

	// after update leader's term >= rf.currentTerm
	// 重置选举时间，为下次选举计算时间间隔
	rf.state = Follower
	rf.ResetVoteTimer()

	if args.Term > rf.currentTerm {
		rf.currentTerm = args.Term // 更新item
		rf.votedFor = -1           // 更新记录为未记录(*)
		rf.state = Follower
		rf.persist()
	}

	// log print
	if len(args.Entries) == 0 { // heartBeats RPC
		DPrintf("server %v received leader %v heartbeat, server's lastIncludeIndex=%v, PrevLogIndex=%v, len(entriese)=%v\n",
			rf.me, args.LeaderId, rf.lastIncludedIndex, args.PrevLogIndex, len(args.Entries))
	} else { // AppendEntries RPC
		DPrintf("server %v received leader %v appendEntries, server's lastIncludeIndex=%v, PrevLogIndex=%v, len(entriese)=%v\n",
			rf.me, args.LeaderId, rf.lastIncludedIndex, args.PrevLogIndex, len(args.Entries))
	}

	isConflict := false

	// 2.Reply false if log doesn't contain an entry at prevLogIndex whose term matches prevLogTerm($5.3)
	// 校验 PrevLogIndex 和 PrevLogTerm 是否合法
	if args.PrevLogIndex < rf.lastIncludedIndex {
		// 过时的RPC，leader的prevLogIndex在当前服务器的lastIncludedIndex之前
		reply.Term = rf.currentTerm
		reply.Success = true
		return
	} else if args.PrevLogIndex >= rf.VirtualLogIndex(len(rf.log)) {
		// PrevLogIndex位置不存在日志
		reply.XTerm = -1
		reply.XLen = rf.VirtualLogIndex(len(rf.log)) // 返回信息包括当前server整个log的长度
		isConflict = true
		DPrintf("server %v log in PrevLogIndex: %v doesn't have log item, log's length is %v\n",
			rf.me, args.PrevLogIndex, reply.XLen)
	} else if args.PrevLogTerm != rf.log[rf.RealLogIndex(args.PrevLogIndex)].Term {
		// PrevLogIndex位置存在日志，但term不匹配
		reply.XTerm = rf.log[rf.RealLogIndex(args.PrevLogIndex)].Term
		idx := args.PrevLogIndex
		for idx > rf.commitIndex && rf.log[rf.RealLogIndex(idx)].Term == reply.XTerm {
			idx -= 1
		}
		reply.XIndex = idx + 1
		reply.XLen = rf.VirtualLogIndex(len(rf.log))
		isConflict = true
		DPrintf("server %v log in PrevLogIndex: %v term doesn't match, args.Term=%v, term is %v\n",
			rf.me, args.PrevLogIndex, args.PrevLogTerm, reply.XTerm)
	}

	if isConflict {
		reply.Term = rf.currentTerm
		reply.Success = false
		return
	}

	// 3.if an existing entry conflicts with a new one(same index but different terms),
	//   delete the existing entry and all that follow it($5.3)
	for idx, log := range args.Entries {
		ridx := rf.RealLogIndex(args.PrevLogIndex) + 1 + idx
		if ridx < len(rf.log) && rf.log[ridx].Term != log.Term {
			// 日志发生了冲突，覆盖这个位置开始的所有日志
			rf.log = rf.log[:ridx]
			rf.log = append(rf.log, args.Entries[idx:]...)
			break // 此处没有写break，导致将peer前面的log一直覆盖（bug）
		} else if ridx == len(rf.log) {
			// 没有发生冲突，直接将日志添加
			rf.log = append(rf.log, args.Entries[idx:]...)
			break
		}
	}
	// 4.Append any new entries not already in the log
	if len(args.Entries) != 0 {
		DPrintf("server %v append log successfully, lastApplied=%v, len(log)=%v\n",
			rf.me, rf.lastApplied, len(rf.log))
	}

	rf.persist()

	reply.Success = true
	reply.Term = rf.currentTerm

	if args.LeaderCommit > rf.commitIndex {
		// 5.if leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of last new entry)
		rf.commitIndex = min(rf.VirtualLogIndex(len(rf.log)-1), args.LeaderCommit)
		//if args.LeaderCommit > rf.VirtualLogIndex(len(rf.log)-1) {
		//	rf.commitIndex = rf.VirtualLogIndex(len(rf.log) - 1)
		//} else {
		//	rf.commitIndex = args.LeaderCommit
		//}
		DPrintf("server %v invokes commit gorutine, commitIndex=%v, len(log)=%v\n",
			rf.me, rf.commitIndex, len(rf.log))
		rf.condApply.Signal() // 唤醒检查commit的协程
	}
}

// leader处理appendEntries回复的消息
func (rf *Raft) handleAppendEntries(server int, args *AppendEntriesArgs) {
	// 处理appendEntries的回复消息
	reply := &AppendEntriesReply{}
	ok := rf.sendAppendEntries(server, args, reply)
	if !ok {
		return
	}

	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	// 当前server必须是leader，且term是最新的
	if rf.state != Leader || args.Term != rf.currentTerm {
		return
	}

	// follower或candidate没有崩溃，成功处理了信息
	if reply.Success {
		newMatchIdx := args.PrevLogIndex + len(args.Entries)
		if newMatchIdx > rf.matchIndex[server] {
			rf.matchIndex[server] = newMatchIdx
		}

		rf.nextIndex[server] = rf.matchIndex[server] + 1

		// N是当前的index，用于判断进行commit的时机
		N := rf.VirtualLogIndex(len(rf.log) - 1)
		for N > rf.commitIndex {
			count := 1
			for i := 0; i < len(rf.peers); i++ {
				if i == rf.me {
					continue
				}
				if rf.matchIndex[i] >= N && rf.log[rf.RealLogIndex(N)].Term == rf.currentTerm {
					count += 1
				}
			}
			// 超过半数的server成功更新了操作的log，此时可以commit
			if count > len(rf.peers)/2 {
				break
			}
			N -= 1
		}

		rf.commitIndex = N
		rf.condApply.Signal()

		return
	}

	// follower或candidate的term大于当前leader的，更改leader状态为follower
	if reply.Term > rf.currentTerm {
		rf.currentTerm = reply.Term
		rf.state = Follower
		rf.votedFor = -1
		rf.ResetVoteTimer()
		rf.persist()

		return
	}

	// 发送信息的follower与leader的term一样
	if reply.Term == rf.currentTerm && rf.state == Leader {
		// follower在prevLogIndex处没有与prevLogTerm匹配的Term，导致无法处理
		if reply.XTerm == -1 {
			// 快照长度大于等于follower的日志长度
			DPrintf("leader %v 收到 server %v 的回退请求, 原因是log过短, 回退前的nextIndex[%v]=%v, 回退后的nextIndex[%v]=%v\n", rf.me, server, server, rf.nextIndex[server], server, reply.XLen)
			if rf.lastIncludedIndex >= reply.XLen {
				// 要找的地方已经被snapshot截断，需要InstallSnapshot处理
				rf.nextIndex[server] = rf.lastIncludedIndex
			} else {
				rf.nextIndex[server] = reply.XLen
			}
			return
		}

		i := rf.nextIndex[server] - 1
		if i < rf.lastIncludedIndex {
			i = rf.lastIncludedIndex
		}
		for i > rf.lastIncludedIndex && rf.log[rf.RealLogIndex(i)].Term > reply.XTerm {
			i -= 1
		}

		// 要找的冲突位置已经被snapshot截断
		if i == rf.lastIncludedIndex && rf.log[rf.RealLogIndex(i)].Term > reply.XTerm { // 循环后找到与rf.lastIncludedIndex相等的索引（bug）
			// 要找的地方已经被snapshot截断，需要InstallSnapshot处理
			rf.nextIndex[server] = rf.lastIncludedIndex
		} else if rf.log[rf.RealLogIndex(i)].Term == reply.XTerm {
			//
			DPrintf("leader %v 收到 server %v 的回退请求, 冲突位置的Term为%v, server的这个Term从索引%v开始, 而leader对应的最后一个XTerm索引为%v, 回退前的nextIndex[%v]=%v, 回退后的nextIndex[%v]=%v\n", rf.me, server, reply.XTerm, reply.XIndex, i, server, rf.nextIndex[server], server, i+1)
			rf.nextIndex[server] = i + 1
		} else {
			//
			DPrintf("leader %v 收到 server %v 的回退请求, 冲突位置的Term为%v, server的这个Term从索引%v开始, 而leader对应的XTerm不存在, 回退前的nextIndex[%v]=%v, 回退后的nextIndex[%v]=%v\n", rf.me, server, reply.XTerm, reply.XIndex, server, rf.nextIndex[server], server, reply.XIndex)
			if reply.XIndex <= rf.lastIncludedIndex {
				// 冲突位置被快照截断
				rf.nextIndex[server] = rf.lastIncludedIndex
			} else {
				rf.nextIndex[server] = reply.XIndex
			}
		}

		return
	}
}

// TODO: RequestVote RPCs are initialed by candidate during elections

// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int // candidate's term
	CandidateId  int // candidate requesting vote
	LastLogIndex int // indexes of candidate's last log entry
	LastLogTerm  int // term of candidate's last log entry
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int  // currentTerm for candidate to update itself
	VoteGranted bool // true means candidate received vote
}

// Candidate发送的RequestVote消息

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()

	defer func() {
		rf.mu.Unlock()
	}()

	if args.Term < rf.currentTerm {
		// 1.Reply false if term < currentTerm
		reply.Term = rf.currentTerm
		reply.VoteGranted = false
		DPrintf("server %v rejects to vote for candidate %v : old term: %v args=%+v\n",
			rf.me, args.CandidateId, args.Term, args)
		return
	}

	if args.Term > rf.currentTerm {
		rf.currentTerm = args.Term
		rf.votedFor = -1
		rf.state = Follower
		rf.persist()
	}

	if rf.votedFor == -1 || rf.votedFor == args.CandidateId {
		//2.if votedFor is null or candidateId
		if args.LastLogTerm > rf.log[len(rf.log)-1].Term ||
			(args.LastLogTerm == rf.log[len(rf.log)-1].Term && args.LastLogIndex >= rf.VirtualLogIndex(len(rf.log)-1)) {
			// and candidate's log is at least as up-to-date as receiver's log, grant vote
			rf.currentTerm = args.Term
			rf.votedFor = args.CandidateId
			rf.state = Follower
			rf.ResetVoteTimer()
			rf.persist()

			reply.VoteGranted = true
			DPrintf("server %v votes for candidate %v heartbeat, args=%+v, len(rf.log)=%v\n",
				rf.me, args.CandidateId, args, len(rf.log))
			return
		} else {
			if args.LastLogTerm < rf.log[len(rf.log)-1].Term {
				DPrintf("server %v rejects to vote for candidate %v : older LastLogTerm, args=%+v\n",
					rf.me, args.CandidateId, args)
			} else {
				DPrintf("server %v rejects to vote for candidate %v : shorter log, args=%+v\n",
					rf.me, args.CandidateId, args)
			}
		}
	} else {
		DPrintf("server %v rejects to vote for candidate %v : has voted, args=%+v\n",
			rf.me, args.CandidateId, args)
	}

	reply.Term = rf.currentTerm
	reply.VoteGranted = false
}

// candidate处理RequestVote回复的消息
func (rf *Raft) handlerRequestVote(server int, args *RequestVoteArgs) bool {
	sendArgs := *args
	// 回收follower的投票信息
	reply := RequestVoteReply{}
	ok := rf.sendRequestVote(server, &sendArgs, &reply)
	if !ok {
		return false
	}

	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	// 要保证当前服务器为candidate，且当前时间戳与发送出去信息的时间戳相等
	if rf.state != Candidate || sendArgs.Term != rf.currentTerm {
		return false
	}

	// 如果回复信息的时间戳大于当前时间戳，重置为follower
	if reply.Term > rf.currentTerm {
		rf.currentTerm = reply.Term
		rf.votedFor = -1
		rf.state = Follower
		rf.persist()
	}
	return reply.VoteGranted
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

func (rf *Raft) sendInstallSnapshot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) bool {
	ok := rf.peers[server].Call("Raft.InstallSnapshot", args, reply)
	return ok
}

// leader发送heartBeats消息
func (rf *Raft) SendHeartBeats() {
	DPrintf("leader %v send the heartBeats\n", rf.me)
	for rf.killed() == false {
		<-rf.heartbeatTimer.C
		rf.mu.Lock()

		if rf.state != Leader { // 不是leader不能发heartBeat
			rf.mu.Unlock()
			return
		}

		// 正常发送心跳消息，包含两种RPC，一种AppendEntries RPC（包含log的正常AppendEntries 和 不含log的heartBeats），一种InstallSnapshot RPC
		for i := 0; i < len(rf.peers); i++ {
			if i == rf.me {
				continue
			}
			args := &AppendEntriesArgs{
				Term:         rf.currentTerm,
				LeaderId:     rf.me,
				PrevLogIndex: rf.nextIndex[i] - 1,
				LeaderCommit: rf.commitIndex,
			}

			sendInstallSnapshot := false

			if args.PrevLogIndex < rf.lastIncludedIndex {
				// 表示follower有落后的部分且被截断，改为发送同步心跳，日志与leader不匹配
				DPrintf("leader %v 取消向 server %v 广播新的心跳, 改为发送sendInstallSnapshot, lastIncludedIndex=%v, nextIndex[%v]=%v, args = %+v \n", rf.me, i, rf.lastIncludedIndex, i, rf.nextIndex[i], args)
				sendInstallSnapshot = true
			} else if args.PrevLogIndex < rf.VirtualLogIndex(len(rf.log)-1) {
				// 有新的log需要发送，则发送appendEntries
				DPrintf("leader %v 开始向 server %v 广播新的AppendEntries, lastIncludedIndex=%v, nextIndex[%v]=%v, PrevLogIndex=%v, len(Entries) = %v\n", rf.me, i, rf.lastIncludedIndex, i, rf.nextIndex[i], args.PrevLogIndex, len(args.Entries))
				args.Entries = rf.log[rf.RealLogIndex(args.PrevLogIndex+1):]
			} else {
				// 没有新log发送，发送切片长度为0，表示心跳信息
				DPrintf("leader %v 开始向 server %v 广播新的心跳, lastIncludedIndex=%v, nextIndex[%v]=%v, PrevLogIndex=%v, len(Entries) = %v \n", rf.me, i, rf.lastIncludedIndex, i, rf.nextIndex[i], args.PrevLogIndex, len(args.Entries))
				args.Entries = make([]Entry, 0)
			}

			if sendInstallSnapshot {
				go rf.handleInstallSnapshot(i)
			} else {
				args.PrevLogTerm = rf.log[rf.RealLogIndex(args.PrevLogIndex)].Term
				go rf.handleAppendEntries(i, args)
			}
		}

		rf.mu.Unlock()
		rf.ResetHeartTimer(HeartbeatTimeout)
	}
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	//index := -1
	//term := -1
	//isLeader := true

	// Your code here (2B).
	rf.mu.Lock()
	defer func() {
		rf.ResetHeartTimer(15) // 等一个心跳再解锁
		rf.mu.Unlock()
	}()

	if rf.state != Leader {
		return -1, -1, false
	}
	// 增加新的log entry
	newEntry := &Entry{
		Term:    rf.currentTerm,
		Command: command,
	}
	rf.log = append(rf.log, *newEntry)
	rf.persist()

	return rf.VirtualLogIndex(len(rf.log) - 1), rf.currentTerm, true
}

func (rf *Raft) CommitChecker() {
	DPrintf("server %v commit checker\n", rf.me)
	for rf.killed() == false {
		rf.mu.Lock()

		for rf.commitIndex <= rf.lastApplied {
			rf.condApply.Wait()
		}
		msgBuf := make([]*ApplyMsg, 0, rf.commitIndex-rf.lastApplied)
		tmpApplied := rf.lastApplied

		for rf.commitIndex > tmpApplied {
			tmpApplied += 1
			if tmpApplied <= rf.lastIncludedIndex {
				continue
			}

			if rf.RealLogIndex(tmpApplied) >= len(rf.log) {
				DPrintf("server %v commit index %v is out of range\n", rf.me, tmpApplied)
			}

			msg := &ApplyMsg{
				CommandValid: true,
				Command:      rf.log[rf.RealLogIndex(tmpApplied)].Command,
				CommandIndex: tmpApplied,
				SnapshotTerm: rf.log[rf.RealLogIndex(tmpApplied)].Term,
			}

			msgBuf = append(msgBuf, msg)
		}
		rf.mu.Unlock()

		for _, msg := range msgBuf {
			rf.mu.Lock()
			if msg.CommandIndex != rf.lastApplied+1 {
				rf.mu.Unlock()
				continue
			}
			rf.mu.Unlock()

			rf.applych <- *msg

			rf.mu.Lock()
			if msg.CommandIndex != rf.lastApplied+1 {
				rf.mu.Unlock()
				continue
			}
			rf.lastApplied = msg.CommandIndex
			rf.mu.Unlock()
		}
	}
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) collectVote(server int, args *RequestVoteArgs, muVote *sync.Mutex, voteCount *int) {
	ok := rf.handlerRequestVote(server, args)
	if !ok {
		return
	}

	muVote.Lock()
	if *voteCount > len(rf.peers)/2 {
		muVote.Unlock()
		return
	}

	*voteCount += 1
	if *voteCount > len(rf.peers)/2 {
		rf.mu.Lock()
		// 要保证当前服务器为candidate，且当前时间戳与发送出去信息的时间戳相等
		if rf.state != Candidate || rf.currentTerm != args.Term {
			rf.mu.Unlock()

			muVote.Unlock()
			return
		}
		// 成为leader，重新初始化nextIndex和matchIndex
		rf.state = Leader
		for i := 0; i < len(rf.nextIndex); i++ {
			rf.nextIndex[i] = rf.VirtualLogIndex(len(rf.log))
			rf.matchIndex[i] = rf.lastIncludedIndex
		}
		rf.mu.Unlock()
		// 向其他server发送心跳信息维护自己的权威
		go rf.SendHeartBeats()
	}

	muVote.Unlock()
}

func (rf *Raft) Elect() {
	rf.mu.Lock()
	defer func() {
		rf.mu.Unlock()
	}()

	rf.currentTerm += 1
	rf.state = Candidate
	rf.votedFor = rf.me
	rf.persist()

	voteCount := 1
	var muVote sync.Mutex

	args := &RequestVoteArgs{
		Term:         rf.currentTerm,
		CandidateId:  rf.me,
		LastLogIndex: rf.VirtualLogIndex(len(rf.log) - 1),
		LastLogTerm:  rf.log[len(rf.log)-1].Term,
	}

	for i := 0; i < len(rf.peers); i++ { // 获取除自己以外所有服务器的投票情况
		if i == rf.me {
			continue
		}
		go rf.collectVote(i, args, &muVote, &voteCount)
	}
}

func (rf *Raft) ticker() {
	for rf.killed() == false {
		// Your code here (2A)
		// Check if a leader election should be started.
		// 当 rf的voteTimer计时器超时后会向通道发送一个值，<-通过接收通道内的数据继续执行
		<-rf.voteTimer.C
		rf.mu.Lock()

		if rf.state != Leader {
			go rf.Elect()
		}
		rf.ResetVoteTimer()
		rf.mu.Unlock()

		// pause for a random amount of time between 50 and 350
		// milliseconds.
		//ms := 50 + (rand.Int63() % 300)
		//time.Sleep(time.Duration(ms) * time.Millisecond)
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {

	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.log = make([]Entry, 0)
	rf.log = append(rf.log, Entry{Term: 0}) // 论文中提到日志索引从1开始

	rf.nextIndex = make([]int, len(rf.peers))
	rf.matchIndex = make([]int, len(rf.peers))

	rf.state = Follower
	rf.randNum = rand.New(rand.NewSource(int64(rf.me)))

	rf.applych = applyCh
	rf.condApply = sync.NewCond(&rf.mu)

	rf.voteTimer = time.NewTimer(0)
	rf.heartbeatTimer = time.NewTimer(0)
	rf.ResetVoteTimer()

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.readSnapshot(persister.ReadSnapshot())

	// 读取snapshot中的数据lastIncludedIndex
	for i := 0; i < len(rf.nextIndex); i++ {
		rf.nextIndex[i] = rf.VirtualLogIndex(len(rf.log))
	}

	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.CommitChecker()

	return rf
}
